package server

import (
	"container/list"
	"fmt"
	proto "github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"sync"
	"sync/atomic"
	"time"
)

type EventPushRegConfig struct {
	///
	/// 固定推送频率: 每秒推送的事件数。如: freq等于20时,则每隔50毫秒推送一次, 一次可推送多个消息/事件.
	///
	EventPushFrequency uint16
	Ses                *Session
	SeriesID           uint8
}

type EventPushReg struct {
	eventQueue              chan proto.Message
	cfg                     *EventPushRegConfig
	exit                    chan bool
	bufferedEvents0         *list.List
	bufferedEvents1         *list.List
	incomingQueue           *list.List
	ticker                  *time.Timer
	incomingQueueSwitchLock *sync.Mutex
	bumpLock                *sync.Mutex
	nextSeqID               uint32
	seqIDLock               *sync.Mutex
	cfgLock                 *sync.Mutex
}

func NewEventPushReg(cfg *EventPushRegConfig) *EventPushReg {
	reg := &EventPushReg{
		eventQueue:              make(chan proto.Message, 100),
		cfg:                     cfg,
		exit:                    make(chan bool),
		bufferedEvents0:         new(list.List),
		bufferedEvents1:         new(list.List),
		ticker:                  time.NewTimer(time.Millisecond * time.Duration(1000/cfg.EventPushFrequency)),
		incomingQueueSwitchLock: new(sync.Mutex),
		bumpLock:                new(sync.Mutex),
		seqIDLock:               new(sync.Mutex),
		cfgLock:                 new(sync.Mutex),
	}
	reg.incomingQueue = reg.bufferedEvents0
	go reg.run()
	return reg
}

func (epr *EventPushReg) QueueEvent(evt proto.Message) {
	select {
	case epr.eventQueue <- evt:
	case <-epr.exit:
		logger().Warn("event push stopped, incoming messages dropped.",
			zap.Uint8("series_id", epr.cfg.SeriesID),
			zap.String("remote_addr", epr.cfg.Ses.GetRemoteAddr()),
		)
	}

}

func (epr *EventPushReg) genSeqID() uint16 {
	id := atomic.AddUint32(&epr.nextSeqID, 1)
	if id > 50000 { // reset
		epr.seqIDLock.Lock()
		if atomic.LoadUint32(&epr.nextSeqID) > 5000 {
			epr.nextSeqID = 0
		}
		epr.seqIDLock.Unlock()
	}
	return uint16(id) - 1
}

func (epr *EventPushReg) run() {
	if le := logger().Check(zap.DebugLevel, "started event pushing routine"); le != nil {
		le.Write(
			zap.Uint8("seriesID", epr.cfg.SeriesID),
			zap.Uint16("freq", epr.cfg.EventPushFrequency),
		)
	}

	for {
		select {
		case <-epr.exit:
			return
		case <-epr.ticker.C:
			epr.cfgLock.Lock()
			epr.ticker.Reset(time.Millisecond * time.Duration(1000/epr.cfg.EventPushFrequency))
			epr.cfgLock.Unlock()
			go epr.bump()
		case m := <-epr.eventQueue:
			epr.incomingQueueSwitchLock.Lock()
			(*epr.incomingQueue).PushFront(m)
			epr.incomingQueueSwitchLock.Unlock()
		}
	}
}

// bump bump queued messages ready for sending
func (epr *EventPushReg) bump() {
	epr.bumpLock.Lock()
	defer epr.bumpLock.Unlock()
	var workingQueue *list.List
	epr.incomingQueueSwitchLock.Lock()
	if epr.incomingQueue == epr.bufferedEvents0 {
		epr.incomingQueue = epr.bufferedEvents1
		workingQueue = epr.bufferedEvents0
	} else {
		epr.incomingQueue = epr.bufferedEvents0
		workingQueue = epr.bufferedEvents1
	}

	epr.incomingQueueSwitchLock.Unlock()
	messages := make([]proto.Message, workingQueue.Len())
	for i := 0; ; i++ {
		m := workingQueue.Front()
		if m == nil {
			break
		}
		workingQueue.Remove(m)
		if le := logger().Check(zap.DebugLevel, "preparing message for batch pushing:"); le != nil {
			le.Write(
				zap.String("remote_addr", epr.cfg.Ses.GetRemoteAddr()),
				zap.Uint8("series_id", epr.cfg.SeriesID),
				zap.String("message", fmt.Sprintf("%+v", m.Value)),
			)
		}
		messages[i] = m.Value.(proto.Message)
	}

	if len(messages) < 1 {
		return
	}
	p := NewPush(epr.cfg.SeriesID, epr.genSeqID(), messages)
	_, err := epr.cfg.Ses.conn.SendData(p.Encode())
	if err != nil {
		logger().Warn("failed to push events.", zap.Error(err))
	}
}

func (epr *EventPushReg) SetFrequency(freq uint16) {
	epr.cfgLock.Lock()
	epr.cfg.EventPushFrequency = freq
	epr.cfgLock.Unlock()
}

func (epr *EventPushReg) Stop() {
	select {
	case <-epr.exit:
		return
	default:
	}
	if le := logger().Check(zap.DebugLevel, "stopping event pushing routine"); le != nil {
		le.Write(
			zap.Uint8("seriesID", epr.cfg.SeriesID),
			zap.Uint16("freq", epr.cfg.EventPushFrequency),
		)
	}
	close(epr.exit)
	epr.ticker.Stop()
}
